package com.gome.boot.config.mq;

import com.gome.boot.common.util.CommonUtils;
import com.gome.boot.config.mq.annotation.RocketMQ;
import com.gome.boot.config.mq.rocketmq.*;
import com.gome.rocketmq.client.extension.MessageConsumer;
import com.gome.rocketmq.client.extension.impl.MessageProducerImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.util.StringUtils;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

/**
 * RocketMQ基础配置.
 *
 * @author baoxiufeng
 */
public abstract class BaseRocketMqConfig {
    @Autowired
    protected Environment env;

    /**
     * 构建MQ生产者实例.
     *
     * @param topicKey MQ消息的topic
     * @param tagKey MQ消息的tags
     * @return MQ生产者实例
     */
    protected MessageProducerImpl buildMessageProducer(String topicKey, String tagKey) {
        return buildMessageProducer(topicKey, tagKey, null, MQDeleyLevel.LEVEL_0);
    }

    /**
     * 构建MQ生产者实例.
     *
     * @param topicKey MQ消息的topic
     * @param tagKey MQ消息的tags
     * @param level 消息延迟发送级别
     * @return MQ生产者实例
     */
    protected MessageProducerImpl buildMessageProducer(String topicKey, String tagKey, MQDeleyLevel level) {
        return buildMessageProducer(topicKey, tagKey, null, level);
    }

    /**
     * 构建MQ生产者实例.
     *
     * @param topicKey MQ消息的topic
     * @param tagKey MQ消息的tags
     * @param serializeType 序列化类型
     * @return MQ生产者实例
     */
    protected MessageProducerImpl buildMessageProducer(String topicKey, String tagKey, SerializeType serializeType) {
        return buildMessageProducer(topicKey, tagKey, serializeType, MQDeleyLevel.LEVEL_0);
    }

    /**
     * 构建MQ生产者实例.
     *
     * @param topicKey MQ消息的topic
     * @param tagKey MQ消息的tags
     * @param serializeType 序列化类型
     * @param level 消息延迟发送级别
     * @return MQ生产者实例
     */
    protected MessageProducerImpl buildMessageProducer(String topicKey, String tagKey, SerializeType serializeType, MQDeleyLevel level) {
        return buildMessageProducer(topicKey, tagKey, serializeType, false, level);
    }

    /**
     * 构建MQ生产者实例.
     *
     * @param topicKey MQ消息的topic
     * @param tagKey MQ消息的tags
     * @param serializeType 序列化类型
     * @param orderly 顺序发送标识
     * @return MQ生产者实例
     */
    protected MessageProducerImpl buildMessageProducer(String topicKey, String tagKey, SerializeType serializeType, boolean orderly) {
        return buildMessageProducer(topicKey, tagKey, serializeType, orderly, MQDeleyLevel.LEVEL_0);
    }

    /**
     * 构建MQ生产者实例.
     *
     * @param topicKey MQ消息的topic
     * @param tagKey MQ消息的tags
     * @param serializeType 序列化类型
     * @param orderly 顺序发送标识
     * @param level 消息延迟发送级别
     * @return MQ生产者实例
     */
    protected MessageProducerImpl buildMessageProducer(String topicKey, String tagKey, SerializeType serializeType, boolean orderly, MQDeleyLevel level) {
        MessageProducerExtImpl messageProducer = new MessageProducerExtImpl(orderly);
        messageProducer.setTopic(env.getRequiredProperty(topicKey));
        messageProducer.setTags(env.getRequiredProperty(tagKey));
        messageProducer.setGroup(MQConstants.DIAMOND_GROUP_ROCKETMQ_SERVER);
        messageProducer.setDataId(MQConstants.DIAMOND_DATAID_ROCKETMQ_SERVER);
        messageProducer.setSerializeType(serializeType == null ? SerializeType.FASTJSON : serializeType);
        messageProducer.setLevel(level == null ? MQDeleyLevel.LEVEL_0.getLevel() : level.getLevel());
        return messageProducer;
    }

    /**
     * 构建MQ消费者实例.
     *
     * @param messageConsumer 消息处理对象
     * @return MQ消费者实例
     */
    protected MessagePushConsumerExtImpl buildMessageConsumer(MessageConsumer<? extends Serializable> messageConsumer) {
        RocketMQ config = messageConsumer.getClass().getAnnotation(RocketMQ.class);
        if (config == null) {
            throw new IllegalArgumentException("Cannot find the RocketMQ annotation config.");
        }
        if (CommonUtils.isEmpty(config.topic())) {
            throw new IllegalArgumentException("RocketMq consumer topic config must not be empty.");
        }
        MessagePushConsumerExtImpl messagePushConsumer = new MessagePushConsumerExtImpl();
        messagePushConsumer.setTopic(config.isDiamondKey() ? env.getRequiredProperty(config.topic()) : config.topic());
        messagePushConsumer.setSubExpression(config.isDiamondKey() ? env.getProperty(config.tag(), MQConstants.DEFAULT_CONSUMER_SUB_EXPRESSION) : config.tag());
        messagePushConsumer.setGroup(config.group());
        messagePushConsumer.setDataId(config.dataId());
        messagePushConsumer.setOrderly(config.orderly());
        messagePushConsumer.setSerializeType(config.serializeType() == null ? SerializeType.FASTJSON : config.serializeType());
        List<MessageConsumer> messageConsumers = new ArrayList<>(1);
        messageConsumers.add(messageConsumer);
        messagePushConsumer.setMessageConsumers(messageConsumers);
        return messagePushConsumer;
    }

    /**
     * 构建MQ消费者实例.
     *
     * @param messageConsumer 消息处理对象
     * @param topicKey 订阅消息的topic
     * @return MQ消费者实例
     */
    protected MessagePushConsumerExtImpl buildMessageConsumer(MessageConsumer<? extends Serializable> messageConsumer, String topicKey) {
        return buildMessageConsumer(messageConsumer, false, topicKey, null);
    }

    /**
     * 构建MQ消费者实例.
     *
     * @param messageConsumer 消息处理对象
     * @param topicKey 订阅消息的topic
     * @param tagKey 订阅消息的tag
     * @return MQ消费者实例
     */
    protected MessagePushConsumerExtImpl buildMessageConsumer(MessageConsumer<? extends Serializable> messageConsumer, String topicKey, String tagKey) {
        return buildMessageConsumer(messageConsumer, false, topicKey, tagKey);
    }

    /**
     * 构建MQ消费者实例.
     *
     * @param messageConsumer 消息处理对象
     * @param orderly 是否顺序消费（true：顺序消费、false：并发消费）
     * @param topicKey 订阅消息的topic
     * @param tagKey 订阅消息的tag
     * @return MQ消费者实例
     */
    protected MessagePushConsumerExtImpl buildMessageConsumer(MessageConsumer<? extends Serializable> messageConsumer, boolean orderly, String topicKey, String tagKey) {
        return buildMessageConsumer(MQConstants.DIAMOND_GROUP_ROCKETMQ_SERVER, MQConstants.DIAMOND_DATAID_ROCKETMQ_SERVER, messageConsumer, orderly, topicKey, tagKey);
    }

    /**
     * 构建MQ消费者实例.
     *
     * @param group rocketmq注册zk配置group
     * @param dataId rocketmq注册zk配置dataId
     * @param messageConsumer 消息处理对象
     * @param topicKey 订阅消息的topic
     * @return MQ消费者实例
     */
    protected MessagePushConsumerExtImpl buildMessageConsumer(String group, String dataId, MessageConsumer<? extends Serializable> messageConsumer, String topicKey) {
        return buildMessageConsumer(group, dataId, messageConsumer, false, topicKey, null);
    }

    /**
     * 构建MQ消费者实例.
     *
     * @param group rocketmq注册zk配置group
     * @param dataId rocketmq注册zk配置dataId
     * @param messageConsumer 消息处理对象
     * @param topicKey 订阅消息的topic
     * @param tagKey 订阅消息的tag
     * @return MQ消费者实例
     */
    protected MessagePushConsumerExtImpl buildMessageConsumer(String group, String dataId, MessageConsumer<? extends Serializable> messageConsumer, String topicKey, String tagKey) {
        return buildMessageConsumer(group, dataId, messageConsumer, false, topicKey, tagKey);
    }

    /**
     * 构建MQ消费者实例.
     *
     * @param group rocketmq注册zk配置group
     * @param dataId rocketmq注册zk配置dataId
     * @param messageConsumer 消息处理对象
     * @param orderly 是否顺序消费（true：顺序消费、false：并发消费）
     * @param topicKey 订阅消息的topic
     * @param tagKey 订阅消息的tag
     * @return MQ消费者实例
     */
    protected MessagePushConsumerExtImpl buildMessageConsumer(String group, String dataId, MessageConsumer<? extends Serializable> messageConsumer, boolean orderly, String topicKey, String tagKey) {
        MessagePushConsumerExtImpl messagePushConsumer = new MessagePushConsumerExtImpl();
        messagePushConsumer.setTopic(env.getRequiredProperty(topicKey));
        messagePushConsumer.setSubExpression(tagKey == null ? MQConstants.DEFAULT_CONSUMER_SUB_EXPRESSION : env.getProperty(tagKey, MQConstants.DEFAULT_CONSUMER_SUB_EXPRESSION));
        messagePushConsumer.setGroup(group);
        messagePushConsumer.setDataId(dataId);
        messagePushConsumer.setOrderly(orderly);
        List<MessageConsumer> messageConsumers = new ArrayList<>(1);
        messageConsumers.add(messageConsumer);
        messagePushConsumer.setMessageConsumers(messageConsumers);
        return messagePushConsumer;
    }
}
